package com.atguigu.bigdata.spark.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class SparkStreaming01_WordCount_JAVA {
    public static void main(String[] args) throws InterruptedException {
        // TODO 准备环境
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(3));

        //输入DStream
        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost",9999);

        //可以认为JavaReceiverInputStream中的，每隔一秒，会有个RDD,其中封装了这一秒发送过来的数据
        //RDD的元素类型为String,即一行一行的文本
        //这里 JavaReciverInputStream的泛型类型<String>，其实就代表了它底层的RDD的泛型类型

        //开始对接收到的数据执行计算，使用spark core提供的算子，执行应用在DSteam中即可
        //在底层，实际上是会对DStream中的一个个的RDD，执行应用在DStram上的算子
        //产生的新RDD，会作为新DStream中的RDD
        JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });

        //开始进行flatmap reduceBykey 操作
        JavaPairDStream<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        wordCounts.print();;
        // 1. 启动采集器
        jssc.start();
        // 2. 等待采集器的关闭
        jssc.awaitTermination();
        //jssc.close();

    }
}
